{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "f5da0754",
   "metadata": {},
   "outputs": [],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "from pyspark import SparkContext, SparkConf\n",
    "conf = SparkConf()\n",
    "sc = SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "13f4dd2e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 分区相关函数"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "7a048742",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[('A', 0)], [('B', 1), ('A', 2)], [('C', 3)], [('A', 4), ('B', 5)]]\n",
      "[[('A', 0), ('A', 2), ('A', 4)], [('B', 1), ('C', 3), ('B', 5)]]\n"
     ]
    }
   ],
   "source": [
    "# partitionBy\n",
    "rdd = sc.parallelize([\"A\", \"B\", \"A\", \"C\",\"A\",\"B\"],4)\n",
    "rdd2 = rdd.zipWithIndex()\n",
    "print(rdd2.glom().collect())\n",
    "rdd3 = rdd2.partitionBy(2, lambda x: 0 if x=='A' else 1)\n",
    "print(rdd3.glom().collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "85b45990",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[0, 1], [2, 3, 4], [5, 6], [7, 8, 9], [10, 11], [12, 13, 14], [15, 16], [17, 18, 19]]\n",
      "[[5, 6, 12, 13, 14, 15, 16], [0, 1, 2, 3, 4], [17, 18, 19], [7, 8, 9, 10, 11]]\n",
      "[[], [15, 16, 17, 18, 19], [], [], [2, 3, 4, 10, 11], [], [12, 13, 14], [], [7, 8, 9], [0, 1, 5, 6]]\n"
     ]
    }
   ],
   "source": [
    "# repartition\n",
    "rdd = sc.parallelize(range(20),8)\n",
    "print(rdd.glom().collect())\n",
    "rdd1 = rdd.repartition(4)\n",
    "print(rdd1.glom().collect())\n",
    "rdd2 = rdd.repartition(10)\n",
    "print(rdd2.glom().collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "0a4c21d3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]]\n",
      "[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9], [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]]\n",
      "[[], [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19], [], []]\n"
     ]
    }
   ],
   "source": [
    "# coalesce \n",
    "# coalesce算子缩减分区数，用于大数据集过滤后，提高小数据集的执行效率。\n",
    "rdd = sc.parallelize(range(20), 2)\n",
    "print(rdd.glom().collect())\n",
    "rdd1 = rdd.coalesce(4)\n",
    "print(rdd1.glom().collect())\n",
    "rdd2 = rdd.coalesce(4, shuffle=True)\n",
    "print(rdd2.glom().collect())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "eec85470",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 缓存 cache\n",
    "rdd.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2e502e15",
   "metadata": {},
   "outputs": [],
   "source": [
    "# checkpoint\n",
    "sc.setCheckpointDir(\"path\")\n",
    "\n",
    "rdd.checkpoint()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d1bf7abe",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 广播变量\n",
    "# bVar = sc.broadcast(value)\n",
    "# bVar.value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "72bfef60",
   "metadata": {},
   "outputs": [],
   "source": [
    "# 累加器\n",
    "# acc = sc.accumulator(0)\n",
    "# acc.add(1)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
